package com.suxin.module.http;

import com.suxin.module.MyOpenTSDBConfig;
import org.apache.http.HeaderElement;
import org.apache.http.HeaderElementIterator;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.conn.ConnectionKeepAliveStrategy;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHeaderElementIterator;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.protocol.HTTP;
import org.apache.http.protocol.HttpContext;

import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Classname MyHttpClientFactory
 * @Description [ 创建HttpClient ]
 * @Author Tang
 * @Date 2020/10/15 15:12
 * @Created by ASUS
 */
public class MyHttpClientFactory {

    private static final AtomicInteger NUM = new AtomicInteger();

    public static MyHttpClient clientHttpClient(MyOpenTSDBConfig myOpenTSDBConfig) throws Exception {

        Objects.requireNonNull(myOpenTSDBConfig);

        ConnectingIOReactor ioReactor = initIOReactorConfig();

        PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);

        RequestConfig requestConfig = initRequestConfig(myOpenTSDBConfig);

        CloseableHttpAsyncClient httpClient = createPoolingHttpClient(requestConfig, connManager, myOpenTSDBConfig);

        return new MyHttpClient(myOpenTSDBConfig,httpClient,initFixedCycleCloseConnection(connManager));
    }

    /**
     * 后台定时任务
     * @param connManager
     * @return
     */
    private static ScheduledExecutorService initFixedCycleCloseConnection(
            PoolingNHttpClientConnectionManager connManager) {

        ScheduledExecutorService executorService = Executors
                .newSingleThreadScheduledExecutor((r) -> {
                    Thread thread = new Thread(r);
                    thread.setName("Fixed-Cycle-Close-Connection-" + NUM.incrementAndGet());
                    thread.setDaemon(true);
                    return thread;
                });

        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    // 关闭30秒内不活动的链接
                    connManager.closeExpiredConnections();
                    connManager.closeIdleConnections(30, TimeUnit.SECONDS);
                } catch (Exception ex) {

                }
            }
        }, 30, 30, TimeUnit.SECONDS);

        return executorService;
    }

    /**
     * 创建cpu核说的IO线程
     * @return
     * @throws IOReactorException
     */
    private static ConnectingIOReactor initIOReactorConfig() throws IOReactorException {
        IOReactorConfig ioReactorConfig =  IOReactorConfig.custom()
                .setIoThreadCount(Runtime.getRuntime()
                        .availableProcessors())
                .build();
        ConnectingIOReactor ioReactor;
        ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
        return ioReactor;
    }

    /**
     * 自定义RequestConfig
     * @param config
     * @return
     */
    private static RequestConfig initRequestConfig(MyOpenTSDBConfig config) {

        return RequestConfig.custom()
                // ConnectTimeout:连接超时.连接建立时间，三次握手完成时间.
                .setConnectTimeout(config.getHttpConnectTimeout() * 1000)
                // SocketTimeout:Socket请求超时.数据传输过程中数据包之间间隔的最大时间.
                .setSocketTimeout(config.getHttpConnectTimeout() * 1000)
                // ConnectionRequestTimeout:httpclient使用连接池来管理连接，这个时间就是从连接池获取连接的超时时间
                .setConnectionRequestTimeout(config.getHttpConnectTimeout() * 1000)
                .build();

    }

    private static CloseableHttpAsyncClient createPoolingHttpClient(RequestConfig config,
                                                                    PoolingNHttpClientConnectionManager cm,
                                                                    MyOpenTSDBConfig openTSDBConfig
    ) {
        cm.setMaxTotal(100);
        cm.setDefaultMaxPerRoute(100);
        HttpAsyncClientBuilder httpAsyncClientBuilder = HttpAsyncClients.custom()
                .setConnectionManager(cm)
                .setDefaultRequestConfig(config);
        // 如果不是只读，则设置为长连接
        if (!openTSDBConfig.isReadonly()) {
            httpAsyncClientBuilder.setKeepAliveStrategy(myStrategy());
        }
        CloseableHttpAsyncClient client = httpAsyncClientBuilder.build();
        return client;
    }

    private static ConnectionKeepAliveStrategy myStrategy() {
        ConnectionKeepAliveStrategy myStrategy = new ConnectionKeepAliveStrategy() {
            @Override
            public long getKeepAliveDuration(HttpResponse response, HttpContext context) {
                HeaderElementIterator it = new BasicHeaderElementIterator
                        (response.headerIterator(HTTP.CONN_KEEP_ALIVE));
                while (it.hasNext()) {
                    HeaderElement he = it.nextElement();
                    String param = he.getName();
                    String value = he.getValue();
                    if (value != null && param.equalsIgnoreCase
                            ("timeout")) {
                        return Long.parseLong(value) * 1000;
                    }
                }
                return 60 * 1000;//如果没有约定，则默认定义时长为60s
            }
        };
        return myStrategy;
    }


}